//package com.wfm.files.bio;
//
//import org.apache.commons.lang.StringUtils;
//import org.apache.iotdb.session.pool.SessionPool;
//import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
//import org.apache.iotdb.tsfile.write.record.Tablet;
//import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
//import java.io.*;
//import java.nio.MappedByteBuffer;
//import java.nio.channels.FileChannel;
//import java.text.SimpleDateFormat;
//import java.util.*;
//import java.util.concurrent.CountDownLatch;
//import java.util.concurrent.atomic.AtomicLong;
//import java.util.stream.Collectors;
//
//public class CSV2Iotdb_MultiThread {
//    public static final String deviceId = "root.ylyx.all";
//    public AtomicLong index = new AtomicLong(0);
//    public long size;
//    public int buffer_size;
//    public RandomAccessFile orign_raf;
//    public FileChannel orignFileChannel;
//    public static volatile TreeMap<Long, String> treeMap = new TreeMap<>(new Comparator<Long>() {
//        @Override
//        public int compare(Long o1, Long o2) {
//            return o1 > o2 ? 1 : -1;
//        }
//    });
//
//    public CSV2Iotdb_MultiThread(File orign, int buffer_size) throws FileNotFoundException {
//        RandomAccessFile orign_raf = new RandomAccessFile(orign, "rw");
//        this.orignFileChannel = orign_raf.getChannel();
//        this.buffer_size = buffer_size;
//        this.size = orign.length();
//    }
//
//    public static void main(String[] args) throws Exception {
//        if (args.length < 3) {
//            System.out.println("use command as:");
//            System.out.println("java -jar  CSV2Redis-1.0-SNAPSHOT.jar /dir/csv2iotdb.properties /dir/csvDir /dir/log.txt thread_num(IntegerValue) buffersize(M)");
//            return;
//        }
//        int num = 1;
//        if (args.length > 3) {
//            try {
//                num = Integer.parseInt(args[3]);
//            } catch (NumberFormatException e) {
//                System.out.println("ERROR! thread nums should be Integer Value !");
//            }
//            System.out.println("workThread nums set " + num + "");
//        } else {
//            System.out.println("workThread nums non-essential default 1");
//        }
//
//        int bufferSize = 5;
//        if (args.length > 4) {
//            try {
//                bufferSize = Integer.parseInt(args[4]);
//            } catch (NumberFormatException e) {
//                System.out.println("ERROR! bufferSize should be Integer Value !");
//            }
//            System.out.println("bufferSize set " + bufferSize + "MB");
//        } else {
//            System.out.println("bufferSize non-essential default 5MB");
//        }
//
//        Properties propFromFile = new Properties();
//        InputStream inputStream = new FileInputStream(new File(args[0]));
//        propFromFile.load(inputStream);
//        String userName = propFromFile.get("iotdb.usrname").toString();
//        String password = propFromFile.get("iotdb.password").toString();
//        String ip = propFromFile.get("iotdb.ip").toString();
//        String port = propFromFile.get("iotdb.port").toString();
//        String maxsize = propFromFile.get("iotdb.maxsize").toString();
//        System.out.println("establish connection to IotDB...");
//        System.out.println("userName:" + userName);
//        System.out.println("password:" + password);
//        System.out.println("ip:" + ip);
//        System.out.println("port:" + port);
//        System.out.println("maxsize:" + maxsize);
//
//        SessionPool sessionPool = new SessionPool(ip, Integer.parseInt(port), userName, password, Integer.parseInt(maxsize));
//        System.out.println("establish connection to IotDB succeed!");
//        File dir = new File(args[1]);
//        if (dir == null) {
//            System.out.println("dir is not exists !!!");
//            return;
//        }
//        File[] files = dir.listFiles();
//        if (files.length == 0) {
//            System.out.println("dir has no file !!!");
//            return;
//        }
//        File logFile = new File(args[2]);
//        System.out.println("begin..." + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
//        Long file_index = new Long(0);
//        for (File orign : dir.listFiles()) {
//            CountDownLatch countDownLatch = new CountDownLatch(num);
//            System.out.println("读取到了" + orign.getName() + ",当前时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
//            CSV2Iotdb_MultiThread pf = new CSV2Iotdb_MultiThread(orign, bufferSize * 1024 * 1024);
//            for (int i = 0; i < num; i++) {
//                new Thread(pf.new ReadTask(countDownLatch, sessionPool, logFile, file_index)).start();
//            }
//            file_index++;
//            countDownLatch.await();
//        }
//
//        if (treeMap != null && treeMap.size() > 0) {
//            System.out.println("后期处理失败的行begin..." + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
//            String tempStr = StringUtils.EMPTY;
//            int temp = 0;
//            //创建空容器
//            List<Map<String, Map<Long, Double>>> containerList = new ArrayList<>();
//            for (Map.Entry test : treeMap.entrySet()) {
//                if (temp == 0) {
//                    tempStr += test.getValue();
//                    temp += 1;
//                } else {
//                    tempStr += test.getValue();
//                    String[] row = tempStr.trim().split(",");
//                    if ("ts".equals(row[1].trim().replace("\"", ""))) {
//                        return;
//                    }
//                    String tag_no = row[2].replaceAll("\"", "");
//                    Long timestamp = Long.valueOf(row[5]);
//                    Double tag_value = Double.parseDouble(row[6]);
//
//                    List<String> tagNoList = containerList.stream().map(s -> s.keySet().iterator().next()).collect(Collectors.toList());
//                    if (tagNoList.contains(tag_no)) {
//                        for (Map<String, Map<Long, Double>> tempContainer : containerList) {
//                            String next = tempContainer.keySet().iterator().next();
//                            if (!StringUtils.equals(next, tag_no)) {
//                                continue;
//                            }
//                            Map<Long, Double> value = tempContainer.get(tag_no);
//                            if (value != null && !value.isEmpty()) {
//                                value.put(timestamp, tag_value);
//                                tempContainer.put(tag_no, value);
//                            } else {
//                                Map<String, Map<Long, Double>> container = new HashMap<>();
//                                Map<Long, Double> longDoubleMap = new HashMap<>();
//                                longDoubleMap.put(timestamp, tag_value);
//                                container.put(tag_no, longDoubleMap);
//                                containerList.add(container);
//                            }
//                        }
//                    } else {
//                        Map<String, Map<Long, Double>> container = new HashMap<>();
//                        Map<Long, Double> longDoubleMap = new HashMap<>();
//                        longDoubleMap.put(timestamp, tag_value);
//                        container.put(tag_no, longDoubleMap);
//                        containerList.add(container);
//                    }
//
//                    tempStr = StringUtils.EMPTY;
//                    temp = 0;
//                }
//            }
//            //插入数据,清空容器
//            for (Map<String, Map<Long, Double>> tempContainer : containerList) {
//                try {
//                    for (Map.Entry<String, Map<Long, Double>> entry : tempContainer.entrySet()) {
//                        String key = entry.getKey();
//                        Map<Long, Double> longDoubleMap = entry.getValue();
//                        List<MeasurementSchema> schemaList = new ArrayList<>();
//                        schemaList.add(new MeasurementSchema(key.toString(), TSDataType.DOUBLE));
//                        Tablet tablet = new Tablet(deviceId, schemaList, longDoubleMap.size());
//                        if (key != null && longDoubleMap != null) {
//                            for (Map.Entry<Long, Double> mapEntry : longDoubleMap.entrySet()) {
//                                int rowIndex = tablet.rowSize++;
//                                tablet.addTimestamp(rowIndex, mapEntry.getKey());
//                                tablet.addValue(key.toString(), rowIndex, mapEntry.getValue());
//                                try {
//                                    if (tablet.rowSize == tablet.getMaxRowNumber()) {
//                                        sessionPool.insertTablet(tablet);
//                                        tablet.reset();
//                                        log(logFile, "插入了" + longDoubleMap.size() + "行数据!  " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
//                                    }
//                                } catch (Exception e) {
//                                    log(logFile, "批量插入iotdb操作失败！");
//                                    e.printStackTrace();
//                                }
//                            }
//                        }
//                    }
//                } catch (Exception e) {
//                    log(logFile, "数据处理方法中发生错误！");
//                    e.printStackTrace();
//                }
//            }
//            containerList.clear();
//            System.out.println("后期处理失败的行end！" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
//        }
//        System.out.println("end!" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
//    }
//
//    private static void log(File file, String content) {
//        try (FileOutputStream fop = new FileOutputStream(file, true)) {
//            if (!file.exists()) {
//                file.createNewFile();
//            }
//            byte[] contentInBytes = content.getBytes();
//            fop.write(contentInBytes);
//            fop.write("\n".getBytes());
//            fop.flush();
//            fop.close();
//        } catch (IOException e) {
//            e.printStackTrace();
//        }
//    }
//
//    class ReadTask implements Runnable {
//        private CountDownLatch countDownLatch;
//        private SessionPool sessionPool;
//        private File logFile;
//        private Long i;
//
//        public ReadTask(CountDownLatch countDownLatch, SessionPool sessionPool, File logFile, Long i) {
//            this.countDownLatch = countDownLatch;
//            this.sessionPool = sessionPool;
//            this.logFile = logFile;
//            this.i = i;
//        }
//
//        @Override
//        public void run() {
//            // 这个任务中需要使用cas获取到当前的 index，并且读取index+buffer值，然后将index改为
//            long cur_index;
//            while ((cur_index = index.get()) < size) {
//                long target_index = (cur_index + buffer_size) > size ? size : cur_index + buffer_size;
//                long D_value = target_index - cur_index;
//                if (index.compareAndSet(cur_index, target_index)) {
//                    //创建空容器
//                    List<Map<String, Map<Long, Double>>> containerList = new ArrayList<>();
//
//                    //读数据，添加到容器
//                    long currentIndex = cur_index;
//                    byte[] content = readFile(cur_index, target_index, D_value);
//                    Scanner scan = new Scanner(new ByteArrayInputStream(content)).useDelimiter("\\R");
//                    while (scan.hasNext()) {
//                        // 这里为对读取文本解析的方法
//                        String next = scan.next();
//                        if (next.indexOf("ts") != -1) {
//                            continue;
//                        }
//                        currentIndex++;
//                        //前一半
//                        if (next.indexOf("INST") == -1) {
//                            synchronized (treeMap) {
//                                treeMap.put(i * 300 * 1024 * 1024 * 1024 * 1024 + currentIndex, next);
//                            }
//                            continue;
//                        }
//                        //后一半
//                        if (next.getBytes()[0] != '\"' || next.getBytes()[5] != '-' || next.getBytes()[8] != '-' || next.getBytes()[11] != '\"') {
//                            synchronized (treeMap) {
//                                treeMap.put(i * 300 * 1024 * 1024 * 1024 * 1024 + currentIndex, next);
//                            }
//                            continue;
//                        }
//                        String[] row = next.trim().split(",");
//                        if ("ts".equals(row[1].trim().replace("\"", ""))) {
//                            return;
//                        }
//                        String tag_no = row[2].replaceAll("\"", "");
//                        Long timestamp = Long.valueOf(row[5]);
//                        Double tag_value = Double.parseDouble(row[6]);
//                        List<String> tagNoList = containerList.stream().map(s -> s.keySet().iterator().next()).collect(Collectors.toList());
//                        if (tagNoList.contains(tag_no)) {
//                            for (Map<String, Map<Long, Double>> tempContainer : containerList) {
//                                String next1 = tempContainer.keySet().iterator().next();
//                                if (!StringUtils.equals(next1, tag_no)) {
//                                    continue;
//                                }
//                                Map<Long, Double> value = tempContainer.get(tag_no);
//                                if (value != null && !value.isEmpty()) {
//                                    value.put(timestamp, tag_value);
//                                    tempContainer.put(tag_no, value);
//                                } else {
//                                    Map<String, Map<Long, Double>> container = new HashMap<>();
//                                    Map<Long, Double> longDoubleMap = new HashMap<>();
//                                    longDoubleMap.put(timestamp, tag_value);
//                                    container.put(tag_no, longDoubleMap);
//                                    containerList.add(container);
//                                }
//                            }
//                        } else {
//                            Map<String, Map<Long, Double>> container = new HashMap<>();
//                            Map<Long, Double> longDoubleMap = new HashMap<>();
//                            longDoubleMap.put(timestamp, tag_value);
//                            container.put(tag_no, longDoubleMap);
//                            containerList.add(container);
//                        }
//                    }
//                    scan.close();
//
//                    //插入数据,清空容器
//                    for (Map<String, Map<Long, Double>> tempContainer : containerList) {
//                        try {
//                            for (Map.Entry<String, Map<Long, Double>> entry : tempContainer.entrySet()) {
//                                String key = entry.getKey();
//                                Map<Long, Double> longDoubleMap = entry.getValue();
//                                List<MeasurementSchema> schemaList = new ArrayList<>();
//                                schemaList.add(new MeasurementSchema(key.toString(), TSDataType.DOUBLE));
//                                Tablet tablet = new Tablet(deviceId, schemaList, longDoubleMap.size());
//                                if (key != null && longDoubleMap != null) {
//                                    for (Map.Entry<Long, Double> mapEntry : longDoubleMap.entrySet()) {
//                                        int rowIndex = tablet.rowSize++;
//                                        tablet.addTimestamp(rowIndex, mapEntry.getKey());
//                                        tablet.addValue(key.toString(), rowIndex, mapEntry.getValue());
//                                        try {
//                                            if (tablet.rowSize == tablet.getMaxRowNumber()) {
//                                                sessionPool.insertTablet(tablet);
//                                                tablet.reset();
//                                                log(logFile, "插入了" + longDoubleMap.size() + "行数据!  " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
//                                            }
//                                        } catch (Exception e) {
//                                            log(logFile, "批量插入iotdb操作失败！");
//                                            e.printStackTrace();
//                                        }
//                                    }
//                                }
//                            }
//                        } catch (Exception e) {
//                            log(logFile, "数据处理方法中发生错误！");
//                            e.printStackTrace();
//                        }
//                    }
//                }
//            }
//            countDownLatch.countDown();
//        }
//
//        public byte[] readFile(long orign_index, long target_index, Long D_value) {
//            // 读取文件,使用一个map内存映射进行读取，可以加速读取吧
//            MappedByteBuffer map;
//            byte[] byteArr = new byte[D_value.intValue()];
//            try {
//                map = orignFileChannel.map(FileChannel.MapMode.READ_ONLY, orign_index, D_value);
//                map.get(byteArr, 0, D_value.intValue());
//            } catch (Exception e) {
//                e.printStackTrace();
//            }
//            return byteArr;
//        }
//    }
//}
//
